import asyncio
import datetime
from collections import defaultdict
from typing import List, Dict, Union

import ujson
from fastapi.encoders import jsonable_encoder
from peewee import DoesNotExist

from common.g.g import G_LOGGER, G_SPU_ES_CLIENT
from common.model.mall.spu import SKU, MiddleSKUAttributeValue, SPU, MiddleSPUAttributeKey
from common.model.stock.stock import MallStock, MallStockItem
from common.pkg.str_id import get_short_uuid
from common.utils.depends.common import Page
from common.utils.resp import RC
from common.v.v import DATAORCODE
from enums.stock_enum import StockTypeEnum
from schema.req_schema.spu_req_schema import AddNewSPUModel, AddNewSKUModel, BulkSPUModel, UpdateSPUReqModel
from schema.req_schema.stock_req_schema import AddNewSPUStockModel
from service.base import Service


class SPUAttributeService(Service):

    async def get_attribute_list(self, mall_id, spu_id_list, sku_id_list):
        query = MiddleSKUAttributeValue.select().where(MiddleSKUAttributeValue.mall_id == mall_id,
                                                       MiddleSKUAttributeValue.spu_id.in_(spu_id_list),
                                                       MiddleSKUAttributeValue.sku_id.in_(sku_id_list),
                                                       MiddleSKUAttributeValue.removed == False)
        return await self.mgw.execute(query.dicts())

    async def get_attribute_sorted_info(self, mall_id, spu_id_list):
        query = MiddleSPUAttributeKey.select(MiddleSPUAttributeKey.spu_id,
                                             MiddleSPUAttributeKey.attribute_key_id,
                                             MiddleSPUAttributeKey.attribute_value_list,
                                             MiddleSPUAttributeKey.seq). \
            where(MiddleSPUAttributeKey.mall_id == mall_id,
                  MiddleSPUAttributeKey.spu_id.in_(spu_id_list),
                  MiddleSPUAttributeKey.removed == False)
        return await self.mgw.execute(query.dicts())

    async def bulk_insert_spu_attribute(self, mall_id, spu_id, attribute_key_list: List[int],
                                        attribute_key_to_value_dict: Dict[int, List[int]]):
        spu_attribute_list = []
        for index, value in enumerate(attribute_key_list):
            spu_attribute_list.append(
                dict(mall_id=mall_id, spu_id=spu_id, attribute_key_id=value, seq=index,
                     attribute_value_list=attribute_key_to_value_dict[value]))
        # 创建spu属性
        query = MiddleSPUAttributeKey.insert_many(spu_attribute_list)
        await self.mgw.execute(query)
        return True

    async def delete_by_spu(self, mall_id, spu_id):
        query = MiddleSPUAttributeKey.update(removed=True, version=MiddleSPUAttributeKey.version + 1). \
            where(MiddleSPUAttributeKey.mall_id == mall_id,
                  MiddleSPUAttributeKey.spu_id == spu_id,
                  MiddleSPUAttributeKey.removed == False)
        await self.mgw.execute(query)


def _format_spu_sku_attribute_info(spu_info, sku_info_list, attribute_info_dict, sorted_attribute_keys,
                                   sorted_attribute_values):
    """
        将属性信息按照设置的排序放入sku里面
    :param spu_info: {}
    :param sku_info_list: []
    :param attribute_info_dict: {"sku_id1": [{"attribute_key_id": "", "attribute_value_id: ""}], "sku_id2": [{"attribute_key_id": "", "attribute_value_id: ""}]}
    :param sorted_attribute_keys: [10002, 10003]
    :param sorted_attribute_values: {"10002": [100003, 100001, 100004], "10003": [100007, 100009, 100011]}
    :return:
    """
    key_sorted_dict = {k: i for i, k in enumerate(sorted_attribute_keys)}
    value_sorted_dict = {k: {v: i for i, v in enumerate(vs)} for k, vs in sorted_attribute_values.items()}
    for sku_info in sku_info_list:
        attr_infos = attribute_info_dict.get(sku_info["sku_id"])
        new_attr = sorted(attr_infos, key=lambda k: (
            key_sorted_dict[k['attribute_key_id']], value_sorted_dict[k['attribute_key_id']][k['attribute_value_id']]))
        sku_info["attributes"] = new_attr
    spu_info["sku_list"] = list(sku_info_list)
    return spu_info


class SKUService(Service):

    async def create_new_sku(self, mall_id, spu_id, currency_code, sku_list: List[AddNewSKUModel]):
        sku_attribute_list = []
        default_sku = None
        for sku in sku_list:
            attributes = sku.get("attributes")
            sku_id = await SKU.add_new_instance(self.mgw, mall_id=mall_id, spu_id=spu_id,
                                                currency_code=currency_code, **sku)
            if default_sku is None:
                default_sku = sku_id
            for each in attributes:
                sku_attribute_list.append(dict(mall_id=mall_id,
                                               spu_id=spu_id,
                                               sku_id=sku_id,
                                               attribute_key_id=each.get("attribute_key"),
                                               attribute_value_id=each.get("attribute_value")))
        return default_sku, sku_attribute_list

    async def delete_by_spu(self, mall_id, spu_id):
        condition = {
            "mall_id": mall_id,
            "spu_id": spu_id,
            "removed": False
        }
        update_fields = {"removed": True, "version": SKU.version + 1}
        return await SKU.update_instance(self.mgw, condition=condition, update_fields=update_fields)

    async def get_by_spu(self, mall_id, spu_id, fields=None):
        return await SKU.search_instances(self.mgw, condition={"mall_id": mall_id, "spu_id": spu_id, "removed": False},
                                          fields=fields)

    async def get_detail_by_sku_id_list(self, mall_id, sku_id_list, fields=None):
        fields = SKU._parse_fields(fields) or SKU._all_fields()
        query = SKU.select(*fields).where(SKU.mall_id == mall_id, SKU.sku_id.in_(sku_id_list), SKU.removed == False)
        res = await self.mgw.execute(query.dicts())
        return res

    async def update_onsale_status(self, mall_id, spu_id, sku_id, onsale, old_version):
        condition = {
            "mall_id": mall_id,
            "spu_id": spu_id,
            "sku_id": sku_id,
            "version": old_version,
            "removed": False
        }
        update_row = await SKU.update_instance(self.mgw, condition=condition, update_fields={"onsale": onsale,
                                                                                             "version": old_version + 1})
        return update_row

    async def get_sku_detail(self, mall_id, spu_id, sku_id):
        try:
            res = await SKU.get_instance(self.mgw, condition={
                "mall_id": mall_id,
                "spu_id": spu_id,
                "sku_id": sku_id,
                "removed": False
            })
        except DoesNotExist:
            return False, DATAORCODE(data=None, code=RC.NOT_EXIST, msg="商品不存在")
        except Exception as e:
            G_LOGGER.exception(e)
            return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")
        return True, DATAORCODE(data=dict(res))


class SKUAttributeValueService(Service):

    async def delete_by_sku(self, mall_id, spu_id, sku_id_list: Union[List[int], None] = None):
        condition = [MiddleSKUAttributeValue.mall_id == mall_id, MiddleSKUAttributeValue.spu_id == spu_id,
                     MiddleSKUAttributeValue.removed == False]
        if sku_id_list:
            condition.append(MiddleSKUAttributeValue.sku_id.in_(sku_id_list))
        query = MiddleSKUAttributeValue.update(removed=True, version=MiddleSKUAttributeValue.version + 1).where(
            *condition)
        return await self.mgw.execute(query)


class SPUService(Service):

    async def get_sku_list(self, mall_id, spu_id_or_list):
        if isinstance(spu_id_or_list, int):
            spu_id_or_list = [spu_id_or_list].sort()
        query = SKU.select().where(SKU.mall_id == mall_id, SKU.spu_id.in_(spu_id_or_list), SKU.removed == False)
        return await self.mgw.execute(query.dicts())

    async def get_spu_list(self, mall_id, page_info: Page, name: str = None,
                           spu_type: int = None, code: str = None, onsale: bool = None,
                           price_type: int = None):
        base_query = SPU.select(SPU.mall_id, SPU.outer_spu_id, SPU.spu_id, SPU.thu_images, SPU.onsale, SPU.update_time,
                                SPU.name). \
            where(SPU.mall_id == mall_id, SPU.removed == False)
        if code:
            base_query = base_query.where(code=code)
            count = await self.mgw.count(base_query)
            data = await self.mgw.execute(base_query)
            return list(data), int(count)
        if name:
            base_query = base_query.where(SPU.name.contains(name))
        if spu_type:
            base_query = base_query.where(SPU.spu_type == spu_type)
        if onsale is not None:
            base_query = base_query.where(SPU.onsale == onsale)
        if price_type:
            base_query = base_query.where(SPU.price_type == price_type)
        try:
            data = await self.mgw.execute(base_query.paginate(page_info.page, page_info.page_size).dicts())
            count = await self.mgw.count(base_query)
        except Exception as e:
            G_LOGGER.exception(e)
            return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")
        return True, DATAORCODE(data={"data": list(data), "count":int(count)})

    async def get_spu_detail_with_sku(self, mall_id, spu_id):
        cache_key = f"mall:{mall_id}:spu:detail_with_sku:{spu_id}"
        cache_data = await self.cache.get_cache(cache_key)
        if cache_data:
            return True, DATAORCODE(data=ujson.loads(cache_data))

        spu_as = SPUAttributeService(self.mgw, self.redis)
        try:
            res = await SPU.get_instance(self.mgw,
                                         condition={"mall_id": mall_id, "spu_id": spu_id, "removed": False},
                                         )
            spu_id_list = [res.get("spu_id")]
            sku_list = await self.get_sku_list(mall_id, spu_id_list)
            spu_id_to_sku_dict = defaultdict(list)
            sku_id_to_sku_dict = {}
            sku_id_list = []
            for each in sku_list:
                spu_id = each.pop("spu_id")
                sku_id = each.get("sku_id")
                spu_id_to_sku_dict[spu_id].append(each)
                sku_id_to_sku_dict[f"{spu_id}_{sku_id}"] = each
                sku_id_list.append(each.get("sku_id"))
        except DoesNotExist:
            return False, DATAORCODE(data=None, code=RC.NOT_EXIST, msg="商品不存在")
        try:
            attribute_list, sorted_attribute_info = await asyncio.gather(
                spu_as.get_attribute_list(mall_id, spu_id_list, sku_id_list),
                spu_as.get_attribute_sorted_info(mall_id, spu_id_list), return_exceptions=True)
            # attribute_list = await spu_as.get_attribute_list(mall_id, spu_id_list, sku_id_list)
            sku_id_to_attribute_dict = defaultdict(list)
            for each in attribute_list:
                sku_id_to_attribute_dict[each.get("sku_id")].append(
                    {"attribute_key_id": each.get("attribute_key_id"),
                     "attribute_value_id": each.get(
                         "attribute_value_id")}
                )
            # sorted_attribute_info = await spu_as.get_attribute_sorted_info(mall_id, spu_id_list)
            sorted_attribute_info = sorted(sorted_attribute_info, key=lambda i: i["seq"])
            sorted_attribute_key = []
            sorted_attribute_key_to_value = {}
            for i in sorted_attribute_info:
                sorted_attribute_key.append(i.get("attribute_key_id"))
                sorted_attribute_key_to_value[i.get("attribute_key_id")] = i.get("attribute_value_list")
            res["attribute_key_list"] = sorted_attribute_key
            res["attribute_key_to_value_dict"] = sorted_attribute_key_to_value
            res = _format_spu_sku_attribute_info(res, sku_list, sku_id_to_attribute_dict, sorted_attribute_key,
                                                 sorted_attribute_key_to_value)
            return True, DATAORCODE(data=dict(res))
        except Exception as e:
            G_LOGGER.exception(e)
            return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")

    async def get_spu_detail(self, mall_id, spu_id, fields=None):
        if fields is None:
            fields = ["onsale"]
        try:
            return await SPU.get_instance(self.mgw, condition={"mall_id": mall_id, "spu_id": spu_id, "removed": False},
                                          fields=fields)
        except DoesNotExist:
            return None

    async def create_new_spu(self, mall_id, spu_info: AddNewSPUModel):
        spu_info.code = get_short_uuid(prefix="G-")
        spu_data = spu_info.dict(exclude_none=True)

        sku_list = spu_data.pop("sku_list")
        attribute_key_list = spu_data.pop("attribute_key_list")
        attribute_key_to_value_dict = spu_data.pop("attribute_key_to_value_dict")
        spu_as = SPUAttributeService(self.mgw, self.redis)
        sku_s = SKUService(self.mgw, self.redis)
        async with self.mgw.atomic() as f:
            try:
                # 创建spu
                spu_id = await SPU.add_new_instance(self.mgw, mall_id=mall_id, **spu_data)
                await spu_as.bulk_insert_spu_attribute(mall_id, spu_id, attribute_key_list, attribute_key_to_value_dict)
                default_sku, sku_attribute_list = await sku_s.create_new_sku(mall_id, spu_id, spu_info.currency,
                                                                             sku_list)
                await SPU.update_instance(self.mgw, condition={"spu_id": spu_id},
                                          update_fields={"default_sku_id": default_sku})
                query = MiddleSKUAttributeValue.insert_many(sku_attribute_list)
                await self.mgw.execute(query)

                # 将spu信息放入es
                await G_SPU_ES_CLIENT.create("index_spu", spu_id, body={
                    "id": spu_id,
                    "mall_id": mall_id,
                    "name": spu_data.get("name"),
                    "alias": spu_data.get("alias"),
                    "spu_type": spu_data.get("spu_type"),
                    "onsale": spu_data.get("onsale").value if hasattr(spu_data.get("onsale"), "value") else spu_data.get("onsale"),
                    "start_timestamp": spu_data.get("start_timestamp"),
                    "end_timestamp": spu_data.get("end_timestamp")
                })
            except Exception as e:
                await f.rollback()
                G_LOGGER.exception(e)
                return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")
            return True, DATAORCODE(data={"spu_id": spu_id})

    async def delete_spu(self, mall_id, spu_id, old_version):
        condition = {
            "mall_id": mall_id,
            "spu_id": spu_id,
            "removed": False,
            "version": old_version
        }
        update_fields = {
            "version": old_version + 1,
            "removed": True
        }
        sku_s = SKUService(self.mgw, self.redis)
        spu_as = SPUAttributeService(self.mgw, self.redis)
        ss = StockService(self.mgw, self.redis)
        sku_avs = SKUAttributeValueService(self.mgw, self.redis)
        async with self.mgw.atomic() as f:
            try:
                delete_row = await SPU.update_instance(self.mgw, condition=condition, update_fields=update_fields)
                if delete_row:
                    await sku_s.delete_by_spu(mall_id, spu_id)
                    await spu_as.delete_by_spu(mall_id, spu_id)
                    await ss.delete_by_spu(mall_id, spu_id)
                    await sku_avs.delete_by_sku(mall_id, spu_id)

                    cache_key = f"mall:{mall_id}:spu:detail_with_sku:{spu_id}"
                    await self.cache.delete(cache_key)
            except Exception as e:
                await f.rollback()
                G_LOGGER.exception(e)
                return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")
        return True, DATAORCODE(data=delete_row)

    async def bulk_set_spu_status(self, mall_id, bulk_model: BulkSPUModel):
        try:
            query = SPU.update(onsale=bulk_model.onsale, version=SPU.version + 1).where(SPU.mall_id == mall_id,
                                                                                        SPU.spu_id.in_(
                                                                                            bulk_model.spu_id_list),
                                                                                        SPU.removed == False)
            update_row = await self.mgw.execute(query)
            return True, DATAORCODE(data=update_row)
        except Exception as e:
            G_LOGGER.exception(e)
            return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")

    async def get_detail_by_spu_id_list(self, mall_id, spu_id_list, fields=None):
        fields = SPU._parse_fields(fields) or SPU._all_fields()
        query = SPU.select(*fields).where(SPU.mall_id == mall_id, SPU.spu_id.in_(spu_id_list), SPU.removed == False)
        res = await self.mgw.execute(query.dicts())
        return res

    async def update_sku_onsale_status(self, mall_id, spu_id, sku_id, onsale):
        ss = SKUService(self.mgw, self.redis)
        async with self.mgw.atomic() as f:
            try:
                ok, res = await ss.get_sku_detail(mall_id, spu_id, sku_id)
                if not ok:
                    return ok, res
                update_row = await ss.update_onsale_status(mall_id, spu_id, sku_id, onsale, old_version=res.data["version"])
            except Exception as e:
                await f.rollback()
                G_LOGGER.exception(e)
                return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")
        return True, DATAORCODE(data={"update_row": update_row})

    async def update_spu_info(self, mall_id, spu_id, spu_info: UpdateSPUReqModel):
        data = spu_info.dict(exclude_unset=True)
        if not data:
            return False, DATAORCODE(data=None, code=RC.PARAM_MISSING, msg="缺失更新参数")
        async with self.mgw.atomic() as f:
            try:
                spu = await self.get_spu_detail(mall_id, spu_id, fields=["spu_id", "version"])
                if spu:
                    old_version = spu["version"]
                    condition = {
                        "mall_id": mall_id,
                        "spu_id": spu_id,
                        "version": old_version,
                        "removed": False
                    }
                    data["version"] = SPU.version + 1
                    update_row = await SPU.update_instance(self.mgw, condition=condition, update_fields=data)
                    return True, DATAORCODE(data={"update_row": update_row})
                else:
                    return False, DATAORCODE(data=None, code=RC.NOT_EXIST, msg="商品不存在")
            except Exception as e:
                G_LOGGER.exception(e)
                await f.rollback()
                return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")


class StockService(Service):

    async def get_spu_stock_type_info(self, mall_id, spu_id_or_list):
        if isinstance(spu_id_or_list, int):
            spu_id_or_list = [spu_id_or_list]
        query = MallStock.select(MallStock.stock_id, MallStock.stock_type, MallStock.spu_id).where(
            MallStock.mall_id == mall_id,
            MallStock.spu_id.in_(spu_id_or_list),
            MallStock.removed == False)
        return list(await self.mgw.execute(query.dicts()))

    async def remove_old_stock_info(self, stock_id_list):
        query = MallStock.update(removed=True, version=MallStock.version + 1).where(
            MallStock.stock_id.in_(stock_id_list))
        return await self.mgw.execute(query)

    async def remove_old_stock_item_info(self, stock_id_list):
        query = MallStockItem.update(removed=True, version=MallStockItem.version + 1).where(
            MallStockItem.stock_id.in_(stock_id_list))
        return await self.mgw.execute(query)

    async def delete_by_spu(self, mall_id, spu_id):
        stock_info = await self.get_spu_stock_type_info(mall_id, spu_id)
        if stock_info:
            stock_id_list = [i.get("stock_id") for i in stock_info]
            await self.remove_old_stock_info(stock_id_list)
            await self.remove_old_stock_item_info(stock_id_list)
        return True

    async def get_stock_detail_by_spu(self, mall_id, spu_id):
        ms = MallStock.select(MallStock.stock_id, MallStock.mall_id, MallStock.spu_id, MallStock.stock_type,
                              MallStock.sku_id). \
            where(MallStock.mall_id == mall_id, MallStock.spu_id == spu_id, MallStock.removed == False)
        try:
            ms_data = await self.mgw.execute(ms.dicts())
        except Exception as e:
            G_LOGGER.exception(e)
            return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")

        stock_ids = [i.get("stock_id") for i in ms_data]
        stock_id_to_sku_dict = {i.get("stock_id"): i.get("sku_id") for i in ms_data}
        if stock_ids:
            query = MallStockItem.select(MallStockItem.stock_id, MallStockItem.id, MallStockItem.date,
                                         MallStockItem.start_time, MallStockItem.end_time,
                                         MallStockItem.quantity, MallStockItem.is_special). \
                where(MallStockItem.stock_id.in_(stock_ids), MallStockItem.removed == False)
            try:
                ms_item_data = await self.mgw.execute(query.dicts())
            except Exception as e:
                G_LOGGER.exception(e)
                return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")
            stock_id_to_stock_item_dict = defaultdict(list)
            for each in ms_item_data:
                stock_id_to_stock_item_dict[each.get("stock_id")].append(each)
            sku_id_to_stock_item = {}
            for stock_id, sku_id in stock_id_to_sku_dict.items():
                sku_id_to_stock_item[sku_id] = stock_id_to_stock_item_dict.get(stock_id, [])
            stock_type = ms_data[0].get("stock_type")
            return True, DATAORCODE(data={"stock_info": sku_id_to_stock_item, "stock_type": stock_type})
        return True, DATAORCODE(data={"stock_info": {}, "stock_type": None})

    async def get_stock_detail_by_sku(self, mall_id, spu_id, sku_id):
        query = MallStock.select(MallStock.stock_id, MallStock.stock_type).where(MallStock.mall_id == mall_id,
                                                                                 MallStock.spu_id == spu_id,
                                                                                 MallStock.sku_id == sku_id,
                                                                                 MallStock.removed == False)
        try:
            ms_data = await self.mgw.get(query)
        except DoesNotExist:
            # 当前sku没有库存信息
            return True, DATAORCODE(data={"stock_info": [], "stock_type": None})
        except Exception as e:
            G_LOGGER.exception(e)
            return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")
        query = MallStockItem.select(MallStockItem.stock_id, MallStockItem.id, MallStockItem.start_date,
                                     MallStockItem.end_date, MallStockItem.start_time, MallStockItem.end_time,
                                     MallStockItem.quantity, MallStockItem.is_special). \
            where(MallStockItem.stock_id == ms_data.get("stock_id"), MallStockItem.removed == False)
        try:
            ms_item_data = await self.mgw.execute(query.dicts())
        except Exception as e:
            G_LOGGER.exception(e)
            return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")
        return True, DATAORCODE(data={"stock_info": ms_item_data, "stock_type": ms_data.get("stock_type")})

    def _build_stock_info(self, stock_id, stock_type, stock_info, special):
        if stock_type == StockTypeEnum.total:
            stock_item_list = [{"stock_id": stock_id, "is_special": False, **each} for each in stock_info]
        elif stock_type == StockTypeEnum.date:
            stock_item_list = []
            for each in stock_info:
                start_date = datetime.datetime.strptime(each["start_date"], "%Y-%m-%d")
                end_date = datetime.datetime.strptime(each["end_date"], "%Y-%m-%d")
                while start_date <= end_date:
                    stock_item_list.append({
                        "stock_id": stock_id,
                        "is_special": special,
                        "date": start_date.date(),
                        "quantity": each["quantity"]
                    })
                    start_date += datetime.timedelta(days=1)
        elif stock_type == StockTypeEnum.time:
            stock_item_list = []
            for each in stock_info:
                start_date = datetime.datetime.strptime(each["start_date"], "%Y-%m-%d")
                end_date = datetime.datetime.strptime(each["end_date"], "%Y-%m-%d")
                while start_date <= end_date:
                    stock_item_list.append({
                        "stock_id": stock_id,
                        "is_special": special,
                        "date": start_date.date(),
                        "quantity": each["quantity"],
                        "start_time": each["start_time"],
                        "end_time": each["end_time"]
                    })
                    start_date += datetime.timedelta(days=1)
        else:
            stock_item_list = []
        return stock_item_list

    async def set_stock(self, mall_id, spu_id, stock_model: AddNewSPUStockModel):
        has_old_stock_info = False
        old_stock_id_list = None
        old_stock_info = await self.get_spu_stock_type_info(mall_id, spu_id)
        if old_stock_info:
            has_old_stock_info = True
            old_stock_type = old_stock_info[0].get("stock_type")
            if old_stock_type != stock_model.stock_type:
                return False, DATAORCODE(data=None, code=RC.FORBIDDING, msg="不允许修改商品的库存类型")
            old_stock_id_list = [i.get("stock_id") for i in old_stock_info]
            # 2. 获取spu下所有的sku
        sku_s = SKUService(self.mgw, self.redis)
        sku_list = await sku_s.get_by_spu(mall_id, spu_id)
        if not sku_list:
            return False, DATAORCODE(data=None, code=RC.NOT_EXIST, msg="商品异常")
        sku_id_list = [i.get("sku_id") for i in sku_list]
        stock_data = jsonable_encoder(stock_model.dict(exclude_none=True))
        special_stock_item_list = stock_data.get("special_stock_item_list")
        special_stock_bulk = stock_data.get("special_stock_bulk")
        sku_id_to_special_stock = {}
        if special_stock_item_list:
            if special_stock_bulk is True:
                for sku_id in sku_id_list:
                    sku_id_to_special_stock[sku_id] = special_stock_item_list
            else:
                sku_id_to_special_stock = defaultdict(list)
                for each in special_stock_item_list:
                    sku_id = each.pop("sku_id")
                    if sku_id not in sku_id_list:
                        return False, DATAORCODE(data=None, code=RC.PARAM_ERROR,
                                            msg=f"设置库存失败,当前商品:{spu_id}不存在sku={sku_id}的商品")
                    sku_id_to_special_stock[sku_id].append(each)

        create_stock_base = {"mall_id": mall_id, "spu_id": spu_id, "stock_type": stock_model.stock_type}
        async with self.mgw.atomic() as f:
            try:
                # 1. 删除旧数据
                if has_old_stock_info:
                    await self.remove_old_stock_info(old_stock_id_list)
                    await self.remove_old_stock_item_info(old_stock_id_list)
                # 2. 创建新的sku库存数据
                for sku_id in sku_id_list:
                    create_stock_base["sku_id"] = sku_id
                    query = MallStock.insert(**create_stock_base)
                    stock_id = await self.mgw.execute(query)
                    stock_item_list = self._build_stock_info(stock_id, stock_model.stock_type, stock_data.get("stock_item_list"), special=False)
                    query = MallStockItem.insert_many(stock_item_list)
                    await self.mgw.execute(query)
                    if stock_model.special_stock_item_list:
                        special_info = sku_id_to_special_stock.get(sku_id)
                        if special_info:
                            special_stock_item_list = self._build_stock_info(stock_id, stock_model.stock_type, special_info, special=True)
                            query = MallStockItem.insert_many(special_stock_item_list)
                            await self.mgw.execute(query)
            except Exception as e:
                await f.rollback()
                G_LOGGER.exception(e)
                return False, DATAORCODE(data=None, code=RC.DB_ERROR, msg="服务异常")
            return True, DATAORCODE(data=None)

    async def get_stock(self, mall_id, condition_list):
        """
        condition_list:
        [
            {
                "spu_id": "",
                "sku_id": "",
                "date": "",
                "start_time": "",
                "end_time": ""
            }
        ]
        """
        c = None
        for each in condition_list:
            sub_query = MallStock.select(MallStock.stock_id).where(
                MallStock.mall_id == mall_id,
                MallStock.spu_id == each.pop("spu_id"),
                MallStock.sku_id == each.pop("sku_id"),
                MallStock.removed == False
            )
            if each:
                cd = [getattr(MallStockItem, k) == v for k, v in each.items()]
                cur = [MallStockItem.stock_id.in_(sub_query) & cd & MallStockItem.removed == False]
            else:
                cur = [MallStockItem.stock_id.in_(sub_query) & MallStockItem.removed == False]
            if c:
                c = c | cur
            else:
                c = cur
        query = MallStockItem.select(MallStockItem.date, MallStockItem.start_time, MallStockItem.end_time, MallStockItem.stock_id,
                                     MallStockItem.quantity, MallStock.spu_id, MallStock.mall_id, MallStockItem.prehold_quantity,
                                     MallStock.sku_id).join(MallStock,
                                                            on=(MallStock.stock_id == MallStockItem.stock_id)).where(c)
        return await self.mgw.execute(query.dicts())

    async def prehold_quantity(self, stock_id_to_quantity: dict):
        bulk_query = []
        for stock_id, quantity in stock_id_to_quantity.items():
            query = MallStockItem.update(prehold_quantity=MallStockItem.prehold_quantity+quantity).where(
                MallStockItem.stock_id == stock_id, MallStockItem.removed == False, MallStockItem.prehold_quantity + quantity <= MallStockItem.quantity
            )
            bulk_query.append(query)
        res_list = []
        for q in bulk_query:
            res = await self.mgw.execute(q)
            res_list.append(res)
        return res_list

    async def rollback_quantity(self, stock_id_to_quantity):
        bulk_query = []
        for stock_id, quantity in stock_id_to_quantity.items():
            query = MallStockItem.update(prehold_quantity=MallStockItem.prehold_quantity-quantity).where(
                MallStockItem.stock_id == stock_id, MallStockItem.removed == False
            )
            bulk_query.append(query)
        res_list = []
        for q in bulk_query:
            res = await self.mgw.execute(q)
            res_list.append(res)
        return res_list


